package com.qit.prototype.flink.performance;

import java.util.ArrayList;
import java.util.List;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import com.qit.prototype.flink.entity.Tick;

/**
 * This case test Flink sink output sequence.
 * @author rollgan
 *
 */
public class TestCase7 {
	
	public TestCase7() throws Exception {
		
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		
		List<Tick> ticks = new ArrayList<Tick>();
		
		for (int i = 0; i < 10; i++) {
			Tick tick = new Tick();
			tick.setPrice(1.0);
			tick.setQty(i + 1.0);
			ticks.add(tick);
		}
		
		DataStream<Tick> dsTick = env.fromCollection(ticks);
		
		DataStream<Tick> dsTickFiltered = dsTick.filter(new FilterFunction<Tick>() {
			private static final long serialVersionUID = 2714547514099649866L;

			@Override
			public boolean filter(Tick arg0) throws Exception {
				System.out.println("filter. Thread=" + Thread.currentThread().getName() + ", value=" + arg0.getQty());
				return true;
			}
		});
		
		dsTickFiltered.addSink(new SinkFunction<Tick>() {
			@Override
			public void invoke(Tick value) throws Exception {
				System.out.println("invoke. Thread=" + Thread.currentThread().getName() + ", value=" + value.getQty());
			}
		});
		
		env.execute();
		
	}

}
